import java.sql.*;
import java.time.LocalDateTime;
import java.util.logging.Logger;
import java.util.Map;


class ProduceTask extends Thread {
    private int id;
    private boolean buildMode;
    private int buildModeBatchCnt;
    private int ipMin;
    private int ipMax;
    private Connection conn;
    private PreparedStatement insertStmt;
    private PreparedStatement selectThrTol;
    private PreparedStatement insertHuge;
    private PreparedStatement dropPartStmt;
    
    public boolean initOk;
    
    public ProduceTask(int id, int min, int max) {
        Log.debug("Produce thread " + id + " init start.");
        this.id = id;
        buildMode = Property.buildMode;
        buildModeBatchCnt = 0;
        ipMin = min;
        ipMax = max;
        
        if (createConnection() && creationPreparedStatement()) {
            Log.debug("Produce thread " + id + " init success.");
            initOk = true;
        } else {
            Log.err("Produce thread " + id + " init failed.");
            initOk = false;
        }
    }

    private boolean createConnection() {
        try {
            conn = DriverManager.getConnection(Property.primaryDbURL, Property.username, Property.password);
            Log.debug("Produce thread " + id + " create connection success.");
            return true;
        } catch (SQLException e) {
            Log.debug("Produce thread " + id + " create connection failed.");
            Log.except(e);
            return false;
        }
    }
    
    private void closeConnection() {
        try{
            conn.close();
        } catch (SQLException e) {
            Log.except(e);
        }
    }
    
    private boolean creationPreparedStatement() {
        try {
            insertStmt = conn.prepareStatement(
                "insert into network" +
                    "(host, ip, itime, key1, key2, key3, value, size) " +
                    "values (?, ?, ?, ?, ?, ?, ?, ?);"
            );
            selectThrTol = conn.prepareStatement(
                "select tolerance from network_thr " +
                "where ip=? and key1=? and key2=? and key3=?;"
            );
            insertHuge = conn.prepareStatement(
                "insert into network_huge " +
                    "(ip, key1, key2, key3, itime, value) " + 
                    "values (?, ?, ?, ?, ?, ?);"
            );
            
            if (buildMode) {
                conn.setAutoCommit(false);
            }
            Log.debug("Produce thread " + id + " create prepare statement success.");
            return true;
        } catch (SQLException e) {
            Log.debug("Produce thread " + id + " create prepare statement failed.");
            Log.except(e);
            return false;
        }
    }
    
    private void insertData(NetPkg pkg, Timestamp itime, int value) throws SQLException{
        insertStmt.setString(1, pkg.ip);
        insertStmt.setString(2, pkg.ip);
        insertStmt.setTimestamp(3, itime);
        insertStmt.setString(4, pkg.key1);
        insertStmt.setString(5, pkg.key2);
        insertStmt.setString(6, pkg.key3);
        insertStmt.setInt(7, value);
        insertStmt.setInt(8, DataGen.randomInt(0, 128));
        if (buildMode) { 
            insertStmt.addBatch();
            buildModeBatchCnt++;
        } else {
            insertStmt.execute();
        }
    }
    
    private void insertHugeData(NetPkg pkg, Timestamp itime, int value) throws SQLException {
        if (!buildMode) {
            selectThrTol.setString(1, pkg.ip);
            selectThrTol.setString(2, pkg.key1);
            selectThrTol.setString(3, pkg.key2);
            selectThrTol.setString(4, pkg.key3);
            selectThrTol.execute();
        }

        insertHuge.setString(1, pkg.ip);
        insertHuge.setString(2, pkg.key1);
        insertHuge.setString(3, pkg.key2);
        insertHuge.setString(4, pkg.key3);
        insertHuge.setTimestamp(5, itime);
        insertHuge.setInt(6, value);
        
        if (buildMode) {
            insertHuge.addBatch();
        } else {
            insertHuge.execute();
        }
    }
    
    private void waitRunFlag() {
        while (!ProduceSystem.flagShutdown) {
            if (ProduceSystem.flagRun) {
                break;
            } else {
                try {
                    Thread.sleep(50);
                } catch (InterruptedException e) {
                    Log.except(e);
                }
            }
        }
    }
    
    
    public void run() {
        int value = DataGen.randomInt(0, 5000000);
        int sqlCount = 0;

        Log.debug("Produce thread " + id + " is ready.");
        waitRunFlag();
        while (!ProduceSystem.flagShutdown) {
            NetPkg pkg = DataGen.randomNetPkg(ipMin, ipMax);
            value = (value > 100000000) ? 0 : (value + 1);
            Timestamp itime = DataGen.timestampNow();
            
            try {
                insertData(pkg, itime, value);
                if (sqlCount % 100 == 0) {
                    insertHugeData(pkg, itime, value);
                    sqlCount = 0;
                }
                
                if (buildMode && buildModeBatchCnt > 1000) {
                    buildModeBatchCnt = 0;
                    insertStmt.executeBatch();
                    insertStmt.clearBatch();
                    insertHuge.executeBatch();
                    insertHuge.clearBatch();
                    conn.commit();
                }
                sqlCount++;
                ProduceSystem.tps.countOne();
            } catch (SQLException e) {
                Log.except(e);
            }
        }
        
        closeConnection();
        Log.debug("Produce thread " + id + " exit.");
    }
}


public class ProduceSystem {
    private static int taskNum;
    private static ProduceTask[] tasks;
    
    public static TpsStatistic tps;
    public static boolean flagRun;
    public static boolean flagShutdown;

    public static boolean init() {
        taskNum = Property.produceParallel;
        tasks = new ProduceTask[taskNum];
        tps = new TpsStatistic();
        flagRun = false;

        Log.info("Produce System initing. parallel=" + taskNum);
        int interval = Math.max(Property.pkgTypeNum / taskNum, 1);
        for (int i = 0; i < taskNum; i++) {
            tasks[i] = new ProduceTask(i, (i * interval), ((i + 1) * interval - 1));

            if (!tasks[i].initOk) {
                Log.err("Produce System init failed.");
                shutdown();
                return false;
            }
        }
        Log.info("Produce System init success.");
        return true;
    }

    public static void run() {
        flagRun = false;
        for (int i = 0; i < taskNum; i++) {
            tasks[i].start();
        }
        flagRun = true;
        tps.reset();
        Log.info("Calling all Produce thread, run!");
    }

    public static String statistic() {
        Map<String, Long> res = tps.calculate();

        String str = "[Produce] " +
            "TOTAL:"  + res.get("count").longValue()     + " " + res.get("avg").longValue()     + "/s " +
            "| 5min:" + res.get("count300s").longValue() + " " + res.get("avg300s").longValue() + "/s " +
            "| 1min:" + res.get("count60s").longValue()  + " " + res.get("avg60s").longValue()  + "/s " +
            "| 1s:"   + res.get("speed");
        Log.info(res.get("time") + " " + str);
        return str;
    }

    public static void shutdown(){
        flagShutdown = true;
        Log.info("Produce System shutdown.");
    }
}

